package com.linkstec.szkettle;

import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
/**
 * 用java执行kettle转换，作业
 * 1.通过文件方式执行转换。
 * 2.通过文件方式执行job。
 * 3.执行资源库的中的转换。
 * @author lyl
 *
 */
public class KettleRunTest {
	/**
	 * 1.通过文件方式执行转换。
	 * @param params
	 * @param ktrPath
	 */
	public static void runTransfer(String[] params, String ktrPath) {  
        Trans trans = null;
        try {
            // 初始化
            // 转换元对象
            KettleEnvironment.init();// 初始化
            EnvUtil.environmentInit();
            TransMeta transMeta = new TransMeta(ktrPath);
            // 转换
            trans = new Trans(transMeta);
            
            // 执行转换
            trans.execute(params);
            // 等待转换执行结束
            trans.waitUntilFinished();  
            // 抛出异常  
            if (trans.getErrors() > 0) {  
                throw new Exception(  
                        "There are errors during transformation exception!(传输过程中发生异常)");  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }
	/**
	 * 2.通过文件方式执行job。
	 * @param maps
	 * @param jobPath
	 */
	public static void runJob(Map<String,String> maps, String jobPath) {  
        try {  
            KettleEnvironment.init();  
            // jobname 是Job脚本的路径及名称  
            JobMeta jobMeta = new JobMeta(jobPath, null);
            Job job = new Job(null, jobMeta);
            // 向Job 脚本传递参数，脚本中获取参数值：${参数名}
            // job.setVariable(paraname, paravalue);
            if (null != maps) {
            	Set<Entry<String, String>> set=maps.entrySet();
            	for(Iterator<Entry<String, String>> it=set.iterator();it.hasNext();){
            		Entry<String, String> ent=it.next();
            		job.setVariable(ent.getKey(), ent.getValue());
            	}
            }
            job.start();
            job.waitUntilFinished();
            if (job.getErrors() > 0) {
                throw new Exception(
                        "There are errors during job exception!(执行job发生异常)");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
	/**
	 * 3.执行资源库的中的转换。
	 * @throws KettleException
	 */
	public static void runWithDb() throws KettleException{
        KettleEnvironment.init();
        //创建DB资源库 
        KettleDatabaseRepository repository=new KettleDatabaseRepository(); 
        DatabaseMeta databaseMeta=new DatabaseMeta("130","Oracle","Native(JDBC)","192.168.10.130","orcl","1521","km","km"); 
        //选择资源库 
        KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta=new KettleDatabaseRepositoryMeta("130","kettle","Transformation description",databaseMeta); 
        repository.init(kettleDatabaseRepositoryMeta); 
        //连接资源库
        repository.connect("admin","admin");
        RepositoryDirectoryInterface directoryInterface=repository.loadRepositoryDirectoryTree(); 
        //选择转换
        TransMeta transMeta=repository.loadTransformation("insert_update",directoryInterface,null,true,null); 
        Trans trans=new Trans(transMeta);
        trans.execute(null);
        trans.waitUntilFinished();//等待直到数据结束
        if(trans.getErrors()>0){
            System.out.println("transformation error"); 
        }else{ 
            System.out.println("transformation successfully"); 
        } 
   } 
	public static void main(String[] args) throws KettleException {
//		String[] params = {"1","content","d:\\test1.txt"};
//		String[] params = {};
//        runTransfer(params, "E:\\kettle\\aaaaaa.ktr");
//		runJob(null, "E:\\kettle\\简单表同步_苏州.kjb");
		runWithDb();
    }
}